package com.letv.mq;

import javax.jms.MessageConsumer;
import javax.jms.Session;

import org.apache.qpid.client.AMQAnyDestination;
import org.apache.qpid.client.AMQConnectionFactory;
import org.apache.qpid.client.message.AbstractJMSMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.letv.conf.ConfigFactory;
import com.letv.mq.addMms.DataHandle;

/**
 * 为解决从库同步问题，接收mms-pipline album消息
 */
public class MqPipLineAddMssAlbum {

    /**
     * String
     */
    private final static String MQ_URL = ConfigFactory.getString("mqMssPipline.urlAlbum") + "&sync_publish='all'&sync_ack=true";

    /**
     * String
     */
    private final static String MQ_QUEUE = ConfigFactory.getString("mqMssPipline.queueAlbum");

    /**
     * int, 异常尝试次数
     */
    private final static int EXCEPTION_MAX_NUM = 100;

    /**
     * int, 异常时等待时间。单位ms
     */
    private final static int SEND_INTERVAL_TIME = 5000;

    /**
     * int, 长时间连接失败的等待时间
     */
    private final static int EXCEPTION_LONG_WAIT_TIME = 10 * 60 * 1000;

    /**
     * Logger
     */
    protected final static Logger LOGGER = LoggerFactory.getLogger(MqPipLineAddMssAlbum.class);

    /**
     * 获取MQ服务器的链接
     * @return SessionFactory
     */
    private static SessionFactory getSessionFactory() {
        SessionFactory result = null;
        try {
            AMQConnectionFactory aMQConnectionFactory = new AMQConnectionFactory();
            aMQConnectionFactory.setConnectionURLString(MqPipLineAddMssAlbum.MQ_URL);
            result = new SessionFactory();
            result.setConnectionFactory(aMQConnectionFactory);
        } catch (Exception e) {
            LOGGER.error("MqPipLineAddMss fail to getSessionFactory", e);
        }
        return result;
    }

    /**
     * mq读取线程
     */
    class MQRev implements Runnable {

        /**
         * SessionFactory
         */
        private SessionFactory sessionFactory;

        /**
         * MessageConsumer
         */
        private MessageConsumer messageConsumer;

        /**
         * Session
         */
        private Session session;

        /**
         * logger
         */
        private Logger logger = LoggerFactory.getLogger(MQRev.class);

        protected MQRev(SessionFactory sessionFactory) {
            this.sessionFactory = sessionFactory;
        }

        @Override
        public void run() {
            this.doRev();
        }

        private void confirmConnection() throws Exception {
            if (null == this.sessionFactory) {
                this.sessionFactory = MqPipLineAddMssAlbum.getSessionFactory();
            }
            if (null == this.session) {
                this.session = this.sessionFactory.createSession(false, Session.AUTO_ACKNOWLEDGE);
                this.messageConsumer = this.session.createConsumer(new AMQAnyDestination(MQ_QUEUE));
            }
        }

        /**
         * 处理mq消息
         */
        public void doRev() {
            int exceptionNum = 0;
            while (true) {
                try {
                    this.confirmConnection();
                    AbstractJMSMessage abstractJMSMessage = (AbstractJMSMessage)this.messageConsumer.receive();
                    String messageStr = new String(abstractJMSMessage.toBodyString());
                    DataHandle.dataHandle(messageStr);
//                    logger.info(messageStr);
                    exceptionNum = 0;
                } catch (Exception e) {
                    try {
                        if(exceptionNum++ > EXCEPTION_MAX_NUM) {
                            this.sessionFactory = null;
                            this.session = null;
                            Thread.sleep(EXCEPTION_LONG_WAIT_TIME);
                        } else {
                            Thread.sleep(SEND_INTERVAL_TIME);
                        }
                    } catch (InterruptedException e1) {
                        logger.error("Call doSent thread sleep exception " + e1.getMessage());
                        break;
                    }
                    logger.error("Connection or recieve Error!", e);
                }
            }
        }
    }

    public static void start() {
        for (int i = 0; i < 5; i++) {
            SessionFactory sessionFactory  = MqPipLineAddMssAlbum.getSessionFactory();
            Thread thread = new Thread(new MqPipLineAddMssAlbum().new MQRev(sessionFactory));
            thread.start();
        }
    }

    public static void main(String[] args) {
    }
}